/*
 * This file is a part of qloudgen-esb. 
 * You can redistribute qloudgen-esb and/or modify it under the terms of the Lesser GNU General Public License version 3. 
 * qloudgen-esb is distributed WITHOUT ANY WARRANTY. 
 * 
 * See the Lesser GNU General Public License for more details at http://www.gnu.org/licenses/.
 *  
 * Copyright (C) 2014 Qloudgen Technology
*/

package com.qloudgen.csb;

import java.util.Random;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import org.apache.log4j.Logger;
import org.zeromq.ZContext;
import org.zeromq.ZFrame;
import org.zeromq.ZLoop;
import org.zeromq.ZMQ; 
import org.zeromq.ZMsg;
import org.zeromq.ZLoop.IZLoopHandler;
import org.zeromq.ZMQ.PollItem;
import org.zeromq.ZMQ.Poller;
import org.zeromq.ZMQ.Socket;

public class ZmqProducer implements Runnable {
	private ProducerConfig config;
	private String conn;
    private ZContext ctx;
    private String id;
    private Socket client;
    private Producer<String, byte[]> producer;
    
    private static Logger LOG = Logger.getLogger(ZmqProducer.class.getName());
    
    private Random rnd = new Random();

    public ZmqProducer(ProducerConfig config, ZContext ctx, String conn) {
    	this.config = config;
        this.conn = conn;
        this.ctx = ctx;
        this.id =  CSBUtils.sockId();
    }

    private IZLoopHandler proc = new IZLoopHandler() {
        @Override
        public int handle(ZLoop loop, PollItem item, Object arg) {
            ZMsg msg = ZMsg.recvMsg(client);
            if (msg != null) {
            	ZFrame address = msg.unwrap();
            	String topic = msg.popString();
            	byte[] bmsg = msg.pop().getData();
            	String key = String.valueOf(rnd.nextInt());
            	KeyedMessage<String, byte[]> kmsg = 
            		new KeyedMessage<String, byte[]>(topic, key, bmsg);
            	producer.send(kmsg);
            
            	ZMsg ack = ZMsg.newStringMsg(CSB.WORKER_ACK);
            	ack.wrap(address);
            	ack.send(client);  
            } else {
            	LOG.error(id + ": receiving interrupted.");
            }
        	return 0;
        }
    };
    
    public void run() {
    	producer = new Producer<String, byte[]>(config);    			
        LOG.info(id + ": producer created");
        
        client = ctx.createSocket(ZMQ.REQ);
        client.setIdentity(id.getBytes());
        client.connect(conn);
        LOG.info(id + ": connected to router via " + conn +"...");
        
        ZFrame frame = new ZFrame(CSB.WORKER_READY);
        frame.send(client, 0);

        ZLoop loop = new ZLoop();
    	PollItem poller = new PollItem(client, Poller.POLLIN);
    	loop.addPoller(poller, proc, null);
    	loop.start();
    	loop.destroy();
    	LOG.error(id + ": loop interrupted, exiting...");

    	ctx.destroySocket(client);
        producer.close();
    }    
}

